package com.hch.netty.nacos;

import com.hch.netty.nacos.entity.*;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Time :2021/12/9 16:14
 * @Author: HeCongHui
 */
public class RegisterServer {
    private final int port;


    public RegisterServer(int port) {
        this.port = port;
    }

    private void start() throws InterruptedException {
        // 1.bossGroup 线程用于接收连接，workerGroup 线程用于具体的处理
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // 2.创建服务端启动引导/辅助类：ServerBootstrap
            ServerBootstrap b = new ServerBootstrap();
            // 3.给引导类配置两大线程组,确定了线程模型
            b.group(bossGroup, workerGroup)
                    // (非必备)打印日志
                    .handler(new LoggingHandler(LogLevel.INFO))
                    // 4.指定 IO 模型为 NIO
                    .channel(NioServerSocketChannel.class)
//            服务器端监听数据回调Handler
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) {
                            ChannelPipeline p = ch.pipeline();
                            // 5.可以自定义客户端消息的业务处理逻辑
                            // 新增Marshaling编码器
                            p.addLast(
                                    MarshallingCodeCFactory.buildMarshallingDecoder()
                            );
                            p.addLast(
                                    MarshallingCodeCFactory.buildMarshallingEncoder()
                            );
                            p.addLast(new RegisterHandler());
                        }
                    });
            // 6.绑定端口,调用 sync 方法阻塞直到绑定完成
            ChannelFuture f = b.bind(port).sync();
            System.out.println("网络服务端启动成功");
            //等待服务端监听端口关闭
            // 7.阻塞等待直到服务器Channel关闭 (closeFuture()方法获取Channel 的CloseFuture对象,然后调用sync()方法)
            f.channel().closeFuture().sync();
        } finally {
            // 8.优雅关闭相关线程组资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static final int type_producer = 0;
    public static final int type_consumer = 1;
    /**
     * key 服务的名称 value 多个生产者集群地址列表
     */
    private static final ConcurrentHashMap<String, List<AddressEntity>> keyAddress = new ConcurrentHashMap<>();
    /**
     * key 为连接 value 服务名称
     */
    private static final ConcurrentHashMap<ChannelHandlerContext, String> ctxs = new ConcurrentHashMap<>();

    private class RegisterHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

            System.out.println("链接为:" + ctx);
            AgreementEntity agreementEntity = (AgreementEntity) msg;
            System.out.println(agreementEntity.toString());
            Integer type = agreementEntity.getType();
            switch (type) {
                case type_producer:
                    producerRegister(agreementEntity, ctx);
                    ctxs.put(ctx, agreementEntity.getServiceId());
                    break;
                case type_consumer:
                    consumerRef(agreementEntity, ctx);
                    break;
                default:
                    System.out.println("交易错误");
            }
        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            super.channelReadComplete(ctx);
            ctx.flush();
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            // 如果客户端主动断开连接的时候 走该方法 那个连接主动断开
            //// key 服务的名称 value 多个生产者集群地址列表

            // 服务名称 缓存3个  每个服务名称对应缓存三个集群地址列表
            //  private static ConcurrentHashMap<String, List<AddresEntity>> keyAddres = new ConcurrentHashMap<>();
            // 多个服务名称 对应对个 服务地址集群理列表

            //ChannelHandlerContext ctx==对应地址
            String serviceId = ctxs.get(ctx);
            // 根据服务名称查找缓存地址
            List<AddressEntity> addresEntities = keyAddress.get(serviceId);
            // 遍历缓存地址剔除
            addresEntities.forEach((t) -> {
                if (t.getCtx() == ctx) {
                    //剔除
                    addresEntities.remove(t);
                }
            });
            // 注册中心通知消费者删除失效地址
            // 循环遍历所有的消费者连接，通知  服务名称 那个地址 应该从内存删除
            ctx.close();
        }


    }

    /**
     * 消费者
     *
     * @param agreement
     * @param ctx
     */
    private void consumerRef(AgreementEntity agreement, ChannelHandlerContext ctx) {
        String serviceId = agreement.getServiceId();
        List<AddressEntity> addressEntities = keyAddress.get(serviceId);
        if (addressEntities == null || addressEntities.size() == 0) {
            ctx.writeAndFlush(new RespEntity(500, "未获取接口地址列表"));
            return;
        }
        // 数据格式转换
        ArrayList<AddressDto> addressDtos = new ArrayList<>();
        addressEntities.forEach((t) -> {
            addressDtos.add(new AddressDto(t.getAddress()));
        });
        ctx.writeAndFlush(new RespEntity(200, "接口地址列表", addressDtos));
    }

    /**
     * 生产者
     *
     * @param agreement
     * @param ctx
     */
    private void producerRegister(AgreementEntity agreement, ChannelHandlerContext ctx) {
        String serviceId = agreement.getServiceId();
        // 根据服务名称查找之前是否有缓存地址数据
        List<AddressEntity> listAddress = keyAddress.get(serviceId);
        if (listAddress == null) {
            listAddress = new ArrayList<>();
            keyAddress.put(serviceId, listAddress);

        }
        // 注册中心插入服务地址信息
        String address = agreement.getAddress();
        AddressEntity addressEntity = new AddressEntity(address, ctx);
        listAddress.add(addressEntity);
    }

    public static void main(String[] args) throws InterruptedException {
        new RegisterServer(8009).start();
    }

}
